package com.zf.nospring.rocketmq.trans;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 事务消息监听器。
 * <p>
 * 事务消息共有三种状态，提交状态、回滚状态、中间状态：
 * <p>
 * - LocalTransactionState.COMMIT_MESSAGE: 提交事务，它允许消费者消费此消息。
 * - LocalTransactionState.ROLLBACK_MESSAGE: 回滚事务，它代表该消息将被删除，不允许被消费。
 * - LocalTransactionState.UNKNOW: 中间状态，它代表需要检查消息队列来确定状态。
 */
public class ProducerTransactionListener implements TransactionListener {
    private final ConcurrentHashMap<String, LocalTransactionState> stateMap = new ConcurrentHashMap<>();

    /**
     * 发送完成 半消息 后执行该方法，用来执行本地事务，本地事务成功，则消费者可以消费该消息，否则删除该消息。
     *
     * @return 返回上面提到的三个事务状态之一
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        System.out.println("事务消息：" + new String(msg.getBody()));
        System.out.println("额外参数：" + arg);
        // 本地事务操作
        try {
            // 模拟异常
            // int a = 1 / 0;
            // 事务成功则放入 提交状态
            stateMap.put(msg.getTransactionId(), LocalTransactionState.COMMIT_MESSAGE);
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            // 事务失败则放入 回滚状态
            stateMap.put(msg.getTransactionId(), LocalTransactionState.ROLLBACK_MESSAGE);
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }

    /**
     * 如果 rocketmq "没有收到 上面 executeLocalTransaction 方法的返回 COMMIT_MESSAGE、 ROLLBACK_MESSAGE" 或者 "收到了 UNKNOW " 就执行该方法检查 事务状态，
     * 本地事务成功，则消费者可以消费该消息，否则删除该消息。
     *
     * @return 返回上面提到的三个事务状态之一
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        return stateMap.get(msg.getTransactionId());
    }
}
